package com.chukun.flink.stream.state.key;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author chukun
 * @version 1.0.0
 * @description 带状态的flatmap操作符 valueState配置生存时间的基本使用
 * @createTime 2022年05月15日 22:59:00
 */
public class TTLStateFlatMap extends RichFlatMapFunction<Tuple2<Integer, Integer>, Tuple2<Integer,Integer>> {

    private static final Logger logger = LoggerFactory.getLogger(TTLStateFlatMap.class);

    // ValueState 类型的状态结构
    private transient ValueState<Tuple2<Integer,Integer>> valueState;

    // 标志是 OnCreateAndWrite还是OnReadAndWrite
    private boolean isRead;
    // 标志是采用 ReturnExpiredIfNotCleanUp 还是 NeverReturnExpired
    private boolean isReturn;

    public TTLStateFlatMap(boolean isRead, boolean isReturn) {
        this.isRead = isRead;
        this.isReturn=isReturn;
    }

    @Override
    public void flatMap(Tuple2<Integer, Integer> input, Collector<Tuple2<Integer, Integer>> collector) throws Exception {
        if (input.f1 > 10 && input.f1 < 20) {
            if (isRead) {
                logger.info("不执行操作:{}", input);
                collector.collect(input);
            } else {
                Tuple2<Integer, Integer> currentState = valueState.value();
                logger.info("不执行操作 状态值:{} 输入值:{}", currentState, input);
            }
        } else {
            Tuple2<Integer, Integer> currentState = valueState.value();
            logger.info("currentState before:" + currentState);
            System.out.println("currentStatecurrentState before:" + currentState);
            if (currentState == null) {
                currentState = input;
            } else {
                currentState.f1 = currentState.f1 + input.f1;
            }
            valueState.update(currentState);
            logger.info("currentState after:" + currentState);
            collector.collect(currentState);
        }
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        logger.info("{},{}", Thread.currentThread().getName(), "恢复或初始化状态");

        // 创建一个ValueStateDescriptor状态描述符，指定名称为valueStateFlatMap，类型为 Tuple2<Integer, Integer>
        ValueStateDescriptor<Tuple2<Integer,Integer>> descriptor = new ValueStateDescriptor<>(
                "ValueStateFlatMap",
                TypeInformation.of(new TypeHint<Tuple2<Integer, Integer>>() {}));

        // 设置状态值的生存时间为6秒
        StateTtlConfig.Builder builder = StateTtlConfig.newBuilder(Time.seconds(6));
        if (isRead) {
            builder.setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite);
        } else {
            builder.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite);
        }
        if(isReturn) {
            builder.setStateVisibility(StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp);
        } else {
            builder.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired);
        }
        StateTtlConfig ttlConfig = builder.build();
        // 配置状态的生存时间
        descriptor.enableTimeToLive(ttlConfig);
        valueState = getRuntimeContext().getState(descriptor);

    }

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 封装了数据源函数，以及keyBy操作的数据流
        KeyedStream<Tuple2<Integer, Integer>, Integer> keyedStream = KeyStateBase.before(env);

        // 在keyedStream中使用有状态的FlatMap操作符
        DataStream<Tuple2<Integer, Integer>> resultStream = keyedStream.flatMap(new TTLStateFlatMap(true, false));

        resultStream.print("输出结果 : ");

        env.execute("TTLStateFlatMap");
    }
}
